package com.wolf.etl.core;

import com.wolf.etl.enumeration.DB_TYPE;
import com.wolf.etl.vo.Column;
import com.wolf.etl.vo.ColumnDesc;
import com.wolf.etl.vo.Database;
import com.wolf.etl.vo.Table;
import com.wolf.exception.WolfException;
import com.wolf.util.ReflectionUtil;
import com.wolf.vo.BaseResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by tommy's father on 2017/8/20.
 */
@Service("DatabaseManager")
public class DatabaseManager {

    private static Logger logger = LoggerFactory.getLogger(DatabaseManager.class);

    private static final String logger_prefix = "wolf-etl#com.wolf.etl.core#DatabaseManager";

    private  DataSource dataSource;

    private String datasource_id;

    private Connection conn = null;

    private static Map<String, DatabaseManager> managerMap = new HashMap<String, DatabaseManager>();


    private DatabaseManager() {
    }

    private static final Object lock = new Object();

    private static String getKey(Connection conn) throws SQLException {
        String url = conn.getMetaData().getURL();
        String name = "";
        if (url.contains("oracle")) {
            name = url.substring(url.lastIndexOf(":") + 1);
            return String.format("oracle#%s", name);
        }
        String catalog = conn.getMetaData().getConnection().getCatalog();
        String schema = conn.getMetaData().getConnection().getSchema();
        if (url.contains("mysql")) {
            return String.format("mysql#%s", catalog);
        }
        if (url.contains("sqlserver")) {
            return String.format("sqlserver#%s.%s", catalog, schema);
        }

        return String.format("%s#%s", catalog, schema == null ? "" : schema);
    }

    public static DatabaseManager getInstance(DataSource dataSource) {
        synchronized (lock) {
            DatabaseManager databaseManager = null;
            String key = null;
            try {
                Connection conn = dataSource.getConnection();
                key = getKey(conn);
                databaseManager = managerMap.get(key);

                if (databaseManager == null) {
                    logger.info(String.format("初始化数据库：%s", key));
                    databaseManager = new DatabaseManager();
                    databaseManager.dataSource = dataSource;
                }

                managerMap.put(key, databaseManager);

                conn.close();
                conn = null;
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return databaseManager;
        }
    }

    // 根据数据库连接-类别-模式，生成Database对象。
    public Database newDatabase(String catalogFilter, String schemaFilter) throws SQLException {
        Database database = new Database(catalogFilter, schemaFilter);
        ResultSet rs = null;
        Connection conn = dataSource.getConnection();
        try {
            // 数据库元数据
            DatabaseMetaData dbmd = conn.getMetaData();

        } catch (Exception e) {
        e.printStackTrace();
        }    finally
        {
            try {
                conn.rollback();
                if(!conn.isClosed()) {
                    conn.close();
                    conn = null;
                }
            } catch (Exception e) {
                /* ignore */
            }
        }
        return database;
    }

    public static Database getTable(DataSource dataSource,Database database, String tableName) {

        if (database.getTable(tableName) != null) {
            return database;
        }

        ResultSet rs = null;
        Connection conn = null;
        try {
             conn = dataSource.getConnection();
            DatabaseMetaData dbmd = conn.getMetaData();
            rs = dbmd.getColumns(database.getCatalog(), database.getSchema(), tableName.toUpperCase(), null);
            if(rs == null){
                rs = dbmd.getColumns(database.getCatalog(), database.getSchema(), tableName.toLowerCase(), null);
            }

            // 遍历结果集，取出表的信息，构建Table对象，加入到Databse的Table集合中
            int i=1;
            while (rs.next()) {
                String tn = rs.getString("TABLE_NAME");

                String catalogName = rs.getString("TABLE_CAT");
                String schemaName = rs.getString("TABLE_SCHEM");

                String columnName = rs.getString("COLUMN_NAME");
                int dataType = Integer.parseInt(rs.getString("DATA_TYPE"));
                String columnTypeName = rs.getString("TYPE_NAME");
                Table table = database.getTable(tn);
                if (table == null) {
                    table = new Table(tn);
                    table.setCatalog(catalogName);
                    table.setSchema(schemaName);
                    database.addTable(table);
                }
                table.addColumn(new Column(columnName, dataType,columnTypeName));
                i++;
            }

            // 遍历Table，取出列名和主键信息
            Table table = database.getTable(tableName);
            rs = dbmd.getPrimaryKeys(database.getCatalog(), database.getSchema(),
                    table.getName());
            if (rs.next()) {
                String columnName = rs.getString("COLUMN_NAME");
                table.setPrimaryKey(table.getColumn(columnName));
            }
            rs.close();

        } catch (SQLException e) {
            e.printStackTrace();
        } finally {

                try {
                    if (rs != null) {
                        rs.close();
                    }
                    if(!conn.isClosed()){
                        conn.close();
                        conn = null;
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }

        }
        return database;
    }

    public static final String RESULT_SET = "ResultSet";


    public void init() throws SQLException {
        Connection conn = dataSource.getConnection();
                conn.setAutoCommit(false);
                conn.close();
                conn = null;
    }

    public void commit() throws SQLException {
        Connection conn = dataSource.getConnection();
        conn.commit();
        conn.setAutoCommit(true);
        conn.close();
        conn = null;
    }

    public BaseResult execute(List<String> sqls) {
        BaseResult result = BaseResult.getSuccess();
        Statement stmt = null;

        Connection conn = null;
        try {
             conn = dataSource.getConnection();
            conn.setAutoCommit(false);
            stmt = conn.createStatement();
            for (String sql : sqls) {
                stmt.addBatch(sql);
            }

            int[] rowResults = stmt.executeBatch();
            int index = 0;
            for (int i = 0; i < rowResults.length; i++) {
                if (rowResults[i] == Statement.EXECUTE_FAILED) {
                    throw new SQLException(String.format("Sql[%s] execute failure!", sqls.get(index)));
                }
                index++;
            }

            int[] resultIds = new int[rowResults.length];
            //获取自动生成的ID
//            ResultSet rs = stmt.getGeneratedKeys();
//            index = 0;
//            while (rs.next()) {
//                resultIds[index] = rs.getInt(1);
//                index++;
//            }

            Map<String, Object> results = new HashMap<>();
            results.put(RESULT_SET, resultIds);
            result = BaseResult.getSuccess(results);

            conn.commit();
//            //恢复自动提交模式
//            conn.setAutoCommit(true);
        } catch (SQLException e) {
           e.printStackTrace();
            result = BaseResult.getFailure(e.getMessage());
            try {
                conn.rollback();
            } catch (SQLException e1) {
                e1.printStackTrace();
            }
        } finally {
            try {
                if (null != stmt) {
                    stmt.close();
                    stmt = null;
                }
                if(!conn.isClosed()){
                    conn.close();
                    conn = null;
                }

            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return result;
    }

    private List<String> sqls = new ArrayList<String>();

    public void appendSql(String sql) {
        this.sqls.add(sql);
    }

    public BaseResult execute() {
        return execute(sqls);
    }

    public BaseResult execute(String sql) {
        BaseResult result = BaseResult.getSuccess();
        Statement stmt = null;
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            stmt = conn.createStatement();
            boolean isSucc = false;
            int intRet = stmt.executeUpdate(sql);
            if (intRet > 0) {
                isSucc = true;

                //获取自动生成的ID
                ResultSet rs = stmt.getGeneratedKeys();
                int id = -1;
                if (rs.next()) {
                    id = rs.getInt(1);
                }
                result = BaseResult.getSuccess(id);

                rs.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
            result = BaseResult.getFailure(e.getMessage());
            try {
                conn.rollback();
            } catch (SQLException e1) {
                e1.printStackTrace();
            }
        } finally {
            try {
                if (null != stmt) {
                    stmt.close();
                    stmt = null;
                }
                if(!conn.isClosed()){
                    conn.close();
                    conn = null;
                }

            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return result;
    }

    private DB_TYPE getDbtype(int types) {
        return DB_TYPE.valueOf(types);
    }

    private ColumnDesc buildColumnDesc(ResultSet rs, int index, DB_TYPE dbType) {
        ColumnDesc columnDesc = ColumnDescManager.getConvertDesc(dbType.getTypeValue());
        if (columnDesc == null) {
            throw new WolfException(String.format("Current db type can't convert to column desc.", dbType.getTypeValue()));
        }

        try {
            Method m = ResultSet.class.getMethod(columnDesc.getMethodName(), new Class[]{int.class});
            Class<?> clazz = ReflectionUtil.getClazz(columnDesc.getClassName());
            columnDesc.setNumberType(dbType.isNumberType());
            columnDesc.setDateType(dbType.isDateType());
            columnDesc.setObj(clazz.cast(m.invoke(rs, index)));
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        return columnDesc;
    }

    public List<Map<String, ColumnDesc>> queryResults(String sql) {
        List<Map<String, ColumnDesc>> records = new ArrayList<Map<String, ColumnDesc>>();
        Statement stmt = null;
        Connection conn = null;
        try {
            conn = dataSource.getConnection();

            stmt = conn.createStatement();
            ResultSet rs = stmt.executeQuery(sql);
            ResultSetMetaData resultSetMetaData = rs.getMetaData();
            // 获取结果集的列数
            int index = resultSetMetaData.getColumnCount();
            while (rs.next()) {
                Map<String, ColumnDesc> record = new HashMap<>();
                for (int i = 1; i <= index; i++) {
                    String columnName = resultSetMetaData.getColumnName(i);
                    String columnTypeName = resultSetMetaData.getColumnTypeName(i);
                    DB_TYPE dbType = getDbtype(resultSetMetaData.getColumnType(i));
                    ColumnDesc columnDesc = buildColumnDesc(rs, i, dbType);
                    columnDesc.setColumnTypeName(columnTypeName);
                    record.put(columnName, columnDesc);
                }

                records.add(record);
            }

            rs.close();


        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != stmt) {
                    stmt.close();
                    stmt = null;
                }
                if(!conn.isClosed()){
                    conn.close();
                    conn = null;
                }



            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return records;
    }
}
